use crate::db::Db;
use crate::redis::*;
use crate::util::*;
use anyhow::Result;
use dotenv::var;
use rumqttc::v5::mqttbytes::v5::Filter;
use rumqttc::v5::{AsyncClient, Event, EventLoop, MqttOptions, mqttbytes::QoS};
use serde::{Deserialize, Serialize};
use std::time::Duration;
use tokio_graceful_shutdown::SubsystemHandle;
use tracing::{error, info};

#[derive(Debug, Clone, Serialize, Deserialize)]
enum Command {
    Ping(i64),              // (user_id)
    EnableInput(i64, bool), // (user_id, enable_input)
    Patch(String),
}

pub fn options() -> Result<MqttOptions> {
    let mqtt_host = var("MQTT_HOST")?;
    let mqtt_port: u16 = var("MQTT_PORT")?.parse()?;
    info!("connect to mqtt {}:{}", mqtt_host.clone(), mqtt_port);
    let mut options = MqttOptions::new("cool-course-server", mqtt_host, mqtt_port);
    options.set_keep_alive(Duration::from_secs(5));
    Ok(options)
}

pub async fn run(
    subsys: SubsystemHandle,
    db: Db,
    redis: Redis,
    client: AsyncClient,
    mut eventloop: EventLoop,
) -> Result<()> {
    let _ = client
        .subscribe_many([
            Filter::new(topic_pings(), QoS::AtMostOnce),
            Filter::new(topic_perms(), QoS::ExactlyOnce),
            Filter::new(topic_patch(), QoS::ExactlyOnce),
        ])
        .await?;
    loop {
        tokio::select! {
            _ = subsys.on_shutdown_requested() => {
                info!("mqtt worker stopped.");
                let _ = client.disconnect().await.inspect_err(|e| error!("{}", e));
                break;
            },
            Ok(event) = eventloop.poll() => {
                match event {
                    Event::Incoming(incoming) => {
                        if let Err(e) = proc_incoming(incoming, db.clone(), redis.clone(), &client).await {
                            error!("mqtt process_incoming error -- {}", e);
                        }
                    },
                    _ => {},
                }
            }
        }
    }
    Ok(())
}

pub fn topic_pings() -> String {
    "space:pings".into()
}

pub fn topic_perms() -> String {
    "space:perms".into()
}

pub fn topic_patch() -> String {
    "space:patch".into()
}

pub fn topic_space_perms(space_id: i64) -> String {
    format!("space:perms:{}", space_id)
}

pub fn topic_space_patch(space_id: i64) -> String {
    format!("space:patch:{}", space_id)
}

pub fn topic_state_pings(space_id: i64) -> String {
    format!("space:pings:{}", space_id)
}
